package com.millstein.realtime.util;

import com.millstein.realtime.common.Constants;

/**
 * @Description
 * @Author tsing
 * @Date 2024-10-09 17:02
 */
public class SqlUtil {

    /**
     * 获取kafka-sink中通用的建表语句
     * @param topic 待写入的kafka主题
     * @return
     */
    public static String getKafkaSinkDDL(String topic) {
        return  "with (" +
                "    'connector' = 'kafka'," +
                "    'properties.bootstrap.servers' = '" + Constants.KAFKA_SERVER + "'," +
                "    'topic' = '" + topic + "'," +
                "    'format' = 'json'" +
                ")";
    }

    /**
     * 获取kafka-source中通用的建表语句
     * @param topic 待读取的kafka主题
     * @param groupId 消费者组
     * @return
     */
    public static String getKafkaSourceDDL(String topic, String groupId) {
        return "with (" +
                "    'connector' = 'kafka'," +
                "    'properties.bootstrap.servers' = '" + Constants.KAFKA_SERVER + "'," +
                "    'properties.group.id' = '" + groupId + "'," +
                "    'topic' = '" + topic + "'," +
                "    'format' = 'json'" +
                ")";
    }

    /**
     * 获取upsert-kafka-sink中通用的建表语句
     * @param topic 待写入的kafka主题
     * @return
     */
    public static String getUpsertKafkaSinkDDL(String topic) {
        return "with (" +
                "    'connector' = 'upsert-kafka'," +
                "    'properties.bootstrap.servers' = '" + Constants.KAFKA_SERVER + "'," +
                "    'topic' = '" + topic + "'," +
                "    'key.format' = 'json'," +
                "    'value.format' = 'json'" +
                ")";
    }
}
